package net.xo68.coordinator.core.gid;


import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.entity.Task;
import com.alipay.sofa.jraft.rpc.RpcContext;
import com.alipay.sofa.jraft.rpc.RpcProcessor;
import lombok.extern.slf4j.Slf4j;
import net.xo68.coordinator.core.rpc.ObjectSerialize;
import net.xo68.coordinator.core.rpc.ObjectSerializeFactory;

import java.nio.ByteBuffer;

/**
 * 全局编号处理
 * @author wuxie
 * @version 2020-9-28
 */
@Slf4j
public class NewIdProcessor  implements RpcProcessor<NewIdRequest> {

    private final GidServer gidServer;

    public NewIdProcessor(GidServer gidServer) {
        this.gidServer = gidServer;
    }



    @Override
    public void handleRequest(RpcContext rpcCtx, NewIdRequest request) {

        // 非leader，生成跳转请求
        if (!gidServer.getFsm().isLeader()) {
            rpcCtx.sendResponse(gidServer.redirect());
            return;
        }

        // 构建应答回调
        NewIdResponse response = new NewIdResponse();
        NewIdClosure closure = new NewIdClosure(gidServer, request, response, new Closure() {

            @Override
            public void run(Status status) {
                // 提交后处理
                if (!status.isOk()) {
                    // 提交失败，返回错误信息
                    response.setError(status.getErrorMsg());
                    response.setStatus(false);
                }
                // 成功，返回ValueResponse应答
                rpcCtx.sendResponse(response);

            }
        });

        try {
            // 构建提交任务
            Task task = new Task();
            task.setDone(closure); // 设置回调

            ObjectSerialize objectSerialize = ObjectSerializeFactory.get();
            byte[] bytes=objectSerialize.serialize(request);
            // 填充数据，将请求用 hessian２序列化到 data　字段
            task.setData(ByteBuffer.wrap(bytes));

            // 提交到 raft group
            gidServer.getNode().apply(task);
        } catch (Exception e) {
            // 处理序列化异常
            log.error("Fail to encode IncrementAndGetRequest", e);
            NewIdResponse responseObject = response;
            responseObject.setError(e.getMessage());
            responseObject.setStatus(false);
            rpcCtx.sendResponse(responseObject);
        }
    }

    @Override
    public String interest() {
        return NewIdRequest.class.getName();
    }
}
